package org.apache.ocean.main.replication;

import java.io.DataOutput;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.ocean.main.util.ByteArrayInputStream;

public class ByteBufferPool {
  private ReentrantLock lock = new ReentrantLock();
	private TreeSet<ByteBuffer> available = new TreeSet<ByteBuffer>();
	private Set<ByteBuffer> inuse = new HashSet<ByteBuffer>();
	private int maxCount;
  
	public ByteBufferPool(int initialSize, int count, int maxCount) {
		this.maxCount = maxCount;
		for (int x=0; x < count; x++) {
			get(initialSize);
		}
	}
	
	public static class ByteBuffers {
		private List<ByteBuffer> byteBuffers;
		private int length;
		
		public ByteBuffers(RandomAccessFile randomAccessFile, int bufferSize, int size, ByteBufferPool bufferPool) throws IOException {
			this.length = size;
			int numRead = 0;
			byteBuffers = new ArrayList<ByteBuffer>();
			while (true) {
				if (numRead >= size) break;
				ByteBuffer byteBuffer = bufferPool.get(bufferSize);
				int len = byteBuffer.getBytes().length;
				if ( (len + numRead) > size) {
					len = size - numRead;
				}
				int n = randomAccessFile.read(byteBuffer.getBytes(), 0, len);
				if (n == -1) break;
				byteBuffers.add(byteBuffer);
				numRead += n;
			}
		}
		
		public ByteBuffers(InputStream inputStream, int bufferSize, int size, ByteBufferPool bufferPool) throws IOException {
			this.length = size;
			int numRead = 0;
			byteBuffers = new ArrayList<ByteBuffer>();
			while (true) {
				if (numRead >= size) break;
				ByteBuffer byteBuffer = bufferPool.get(bufferSize);
				int len = byteBuffer.getBytes().length;
				if ( (len + numRead) > size) {
					len = size - numRead;
				}
				int n = inputStream.read(byteBuffer.getBytes(), 0, len);
				if (n == -1) break;
				byteBuffers.add(byteBuffer);
				numRead += n;
			}
		}
		
		public ByteBuffers(ByteBuffer byteBuffer, int length) {
			byteBuffers = new ArrayList<ByteBuffer>(1);
			byteBuffers.add(byteBuffer);
			this.length = length;
		}
		
		public ByteBuffers(List<ByteBuffer> byteBuffers, int length) {
			this.byteBuffers = byteBuffers;
			this.length = length;
		}
    
		public InputStream getInputStream() throws IOException {
			return new ByteArrayInputStream(this);
		}
		
		public void writeTo(DataOutput out) throws IOException {
			int remaining = length;
			for (int i = 0; i < byteBuffers.size(); i++) {
				byte[] buf = byteBuffers.get(i).getBytes();
				int c = Math.min(buf.length, remaining);
				out.write(buf, 0, c);
				remaining -= c;
				if (remaining == 0) {
					break;
				}
			}
		}
		
		public void finished() {
			for (ByteBuffer byteBuffer : byteBuffers) {
				byteBuffer.finished();
			}
		}
		
		public List<ByteBuffer> getByteBuffers() {
			return byteBuffers;
		}

		public int getLength() {
			return length;
		}
	}
	
	public static class ByteBuffer implements Comparable<ByteBuffer> {
		private byte[] bytes;
		private ByteBufferPool byteBufferPool;
		
		public ByteBuffer(int size, ByteBufferPool byteBufferPool) {
			this.bytes = new byte[size];
			this.byteBufferPool = byteBufferPool;
		}
		
		public void finished() {
			byteBufferPool.finished(this);
		}
		
		public int compareTo(ByteBuffer other) {
			return new Integer(size()).compareTo(other.size());
		}
		
		public int size() {
			return bytes.length;
		}
		
		public byte[] getBytes() {
			return bytes;
		}
	}
	
	private void finished(ByteBuffer byteBuffer) {
		lock.lock();
		try {
			inuse.remove(byteBuffer);
			available.add(byteBuffer);
		} finally {
			lock.unlock();
		}
	}
	
	private void checkCount() {
		lock.lock();
		try {
			int dif = available.size() - maxCount;
			if (dif > 0) {
				int count = 0;
				Iterator<ByteBuffer> iterator = available.iterator();
				while (iterator.hasNext() && count < dif) {
					iterator.next();
					iterator.remove();
					count++;
				}
			}
		} finally {
			lock.unlock();
		}
	}
	
	public ByteBuffer get(int size) {
		lock.lock();
		try {
			ByteBuffer byteBuffer = available.last();
			if (byteBuffer == null || size > byteBuffer.size()) {
				byteBuffer = new ByteBuffer(size, this);
			} else {
				available.remove(byteBuffer);
			}
			inuse.add(byteBuffer);
			return byteBuffer;
		} finally {
			lock.unlock();
		}
	}
}
